🔴 Blocking — Correctness: applySerializer is ineffective for scalar SqlParameter values

The CosmosItemSerializer.serialize() contract returns Map<String, Object>, which fundamentally cannot represent scalar values. For custom serializers like BasicCustomItemSerializer, serialize(scalar) calls customMapper.convertValue(scalar, Map.class), which always throws for non-Map-able types (String, Integer, Instant, etc.). The catch(Throwable) block silently swallows the exception, and the parameter retains its default-serialized representation.

Concrete failure trace for this test:

1. Document created with BasicCustomItemSerializer (WRITE_DATES_AS_TIMESTAMPS=false) → createdAt stored as ISO string "2026-03-15T10:30:00Z"
2. SqlParameter("@createdAt", Instant) → setValue() stores rawValue=Instant and default-serializes to epoch number 1773837000 (SDK's ObjectMapper has WRITE_DATES_AS_TIMESTAMPS=true)
3. Clone (line 78): getValue(Object.class) reads the epoch number from JSON property bag → clone's rawValue = Long(1773837000) (type lost)
4. applySerializer(BasicCustomItemSerializer) → serialize(Long) → convertValue(Long, Map.class) → throws IllegalArgumentException
5. catch(Throwable) fires → parameter keeps epoch number 1773837000
6. Query: c.createdAt = @createdAt compares string "2026-03-15T10:30:00Z" vs number 1773837000 → 0 results
7. assertThat(results).hasSize(1) → test fails

This means Bug 2's fix (SqlParameter custom serialization) is a no-op for the exact scenario described in issue #45521 — scalar query parameters with types where the custom serializer changes the wire format.

Root cause: DefaultCosmosItemSerializer.serialize() handles scalars via PrimitiveJsonNodeMap (an internal type), but custom serializers can't — the Map<String, Object> return type makes it impossible to return a scalar from serialize().

Suggested approaches (pick one):

(a) Use the serializer's ObjectMapper directly for scalar parameter values:

void applySerializer(CosmosItemSerializer serializer) {
    if (this.rawValue != null && serializer != null) {
        ObjectMapper mapper = serializer.getItemObjectMapper();
        JsonNode node = mapper.convertValue(this.rawValue, JsonNode.class);
        this.jsonSerializable.getPropertyBag().set("value", node);
    }
}

But this requires BasicCustomItemSerializer (and customer serializers) to call setItemObjectMapper(customMapper) — document this contract.

(b) Add a serializeToNode(T item) method to CosmosItemSerializer with a default implementation using getItemObjectMapper(), and use that in applySerializer instead of serialize().

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

🔴 Blocking — Correctness: Clone via getValue(Object.class) loses the original Java type

This line reconstructs the SqlParameter value by reading from the JSON property bag (getObject("value", Object.class) → getValue(JsonNode)), not from rawValue. The getValue(JsonNode) method converts JSON nodes back to Java primitives: LongNode → Long, DecimalNode → Double, TextNode → String.

For an Instant parameter that was default-serialized as epoch-seconds (a numeric node), getValue(Object.class) returns a Long — the clone's rawValue is Long, not the original Instant. When applySerializer later tries to re-serialize this Long with the custom serializer, it serializes the wrong type.

Fix: Expose rawValue through a package-private accessor and use it for cloning:

// In SqlParameter.java (package-private):
Object getRawValue() { return this.rawValue; }

// Here:
clonedParams.add(new SqlParameter(p.getName(), p.getRawValue()));

Or add a package-private copy constructor:

SqlParameter(SqlParameter original) {
    this.jsonSerializable = new JsonSerializable(
        original.jsonSerializable.getPropertyBag().deepCopy());
    this.rawValue = original.rawValue;
}

This is a prerequisite for any fix to the applySerializer scalar issue — even after serialize() is made to handle scalars, the clone still needs to preserve the original Java type.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

🟡 Recommendation — Error Handling: catch(Throwable) catches fatal JVM errors; log level too low

Two issues with this error handling:

1. Throwable catches Error subclasses — OutOfMemoryError, StackOverflowError, ThreadDeath, and VirtualMachineError are fatal JVM conditions that should never be silently swallowed. The intended use case (Spark connector stubs throwing UnsupportedOperationException) is an Exception, not an Error.

2. LOGGER.debug() masks incorrect query results — When the serializer fails and falls back to default serialization, the parameter value may have a different wire representation than the document field (e.g., epoch number vs ISO string). The query silently returns zero results. At DEBUG level, this is invisible in production. The log message says "does not support serialize()" but the actual failure mode (scalar-to-Map conversion failure) is different and much harder to diagnose.

// Suggested fix:
} catch (Exception e) {
    LOGGER.warn(
        "Custom serializer '{}' failed to serialize SqlParameter value of type '{}'; "
            + "falling back to default serialization. Query results may not match if the "
            + "custom serializer changes the wire format for this type.",
        serializer.getClass().getSimpleName(),
        this.rawValue.getClass().getSimpleName(),
        e);
}

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.